package hasthi.coordinator;

import static hasthi.common.constants.HasthiConstants.ASSIGN_RESOURCE_ACTION;
import static hasthi.common.constants.HasthiConstants.LOGGER_NAME;
import hasthi.common.BaseContext;
import hasthi.common.DistributedLogger;
import hasthi.common.HasthiException;
import hasthi.common.LogFile;
import hasthi.common.RemoteProcessorException;
import hasthi.common.HasthiException.FaultCode;
import hasthi.common.constants.WSDMConstants.SystemState;
import hasthi.container.MessageContext;
import hasthi.container.SimpleMessageContext;
import hasthi.container.UnicastContianier;
import hasthi.instrument.JavaInstrumentation;
import hasthi.manager.AssignResourcesDocument;
import hasthi.manager.AssignResourcesResponseDocument;
import hasthi.manager.ManagerAddress;
import hasthi.manager.ManagerClientUtils;
import hasthi.manager.ManagerContext;
import hasthi.metamodel.SummerizedResourceModel;
import hasthi.metamodel.resource.ManagedResource;
import hasthi.tools.ManagementRingTools;
import hasthi.tools.analyze.DataCollector;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import javax.xml.namespace.QName;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class CoordinationDemon implements Runnable {
    private CoordinatorContext context;
    private List<RemoteManagerContext> managers2remove = new ArrayList<RemoteManagerContext>();
    private List<QName> unAssignResources = new ArrayList<QName>();
    private DistributedLogger dlogger;

    private static Log log = LogFactory.getLog(LOGGER_NAME);

    private DataCollector loc;
    private DataCollector roc;
    private ScheduledFuture demonHandle;
   // private List<QName> resourcesToRemove = new ArrayList<QName>();
    
    public CoordinationDemon(CoordinatorContext context) {
        this.context = context;
        BaseContext baseConfigContext = context.getMyManagerContext().getBaseContext();
        dlogger = baseConfigContext.getDlogger();

        // This code collect statistics about the management system
        LogFile logfile = LogFile.getLog("hasthi_coordinator");
        loc = new DataCollector("loc(c_" + UUID.randomUUID() + ")", logfile,
                DataCollector.WINDOW_SIZE);
        roc = new DataCollector("roc(c_" + UUID.randomUUID() + ")", logfile,
                DataCollector.WINDOW_SIZE);
    }

    public void startDeamon() throws HasthiException {
        if (demonHandle != null) {
            throw new HasthiException("Demon Already started", FaultCode.LocalError);
        }
        int delay = 5000; // delay for 5 sec.
        ManagerContext managerContext = context.getMyManagerContext();
        int period = managerContext.getBaseContext().getCoordinatorEpochSeconds() * 1000;
        log.info("Using " + period + "ms as Coordinator epoch time");
        demonHandle = managerContext.getScheduledExecutorService().scheduleAtFixedRate(this, delay,
                period, TimeUnit.MILLISECONDS);
    }

    public void stopDeamon() throws HasthiException {
        if (demonHandle == null) {
            throw new HasthiException("Demon Already started", FaultCode.LocalError);
        }
        demonHandle.cancel(true);
        demonHandle = null;
    }

    public void run() {

        BaseContext baseConfigContext = context.getMyManagerContext().getBaseContext();
        //SystemStat systemStat = context.getSystemStat();
//        systemStat.reset();
//        systemStat.setManagerCount(context.getManagerCount());
//        systemStat.setResourceCount(context.getResourceCount());
//        systemStat.setMemoryUsage(JavaInstrumentation.getMemoryUsagePercent());
        
        long coordinatorAgeSteps = (System.currentTimeMillis() - context.getMyManagerContext()
                .getCoordinatorTimeStamp())
                / (baseConfigContext.getCoordinatorEpochSeconds() * 1000);
        String msg = "Coordinator " + context.getResourceCount() + " R->"
                + context.getManagerCount() + " Mgrs " + new Date() + " ("
                + JavaInstrumentation.getMemoryUsage() + ") age = " + coordinatorAgeSteps + " "
                + context.getMyManagerContext();
        dlogger.info(msg);
        System.out.println(msg);
        try {

            // at start coordinator send heartbeats often falling back
            // exponentially, then sends it once only
            // 10 epochs
            if (coordinatorAgeSteps == 2 || coordinatorAgeSteps == 4 || coordinatorAgeSteps == 8
                    || (coordinatorAgeSteps > 10 && coordinatorAgeSteps % 10 == 0)) {
                ManagementRingTools.sendCoordinatorHeartbeat(context.getMyManagerContext());
            }

            long point1 = System.nanoTime();
            managers2remove.clear();
            long curretTime = System.currentTimeMillis();

            /**
             * We remove the resource who are repaired or too long since failed or stopped. They are collected by a 
             * model listener in the resource model, and added to resources to remove map
             */
            SummerizedResourceModel resourceModel = context.getResourceModel();
            Map<Long, QName> resource2Remove = context.getResource2Remove();
            synchronized (resource2Remove) {
                Iterator<Long> keys = resource2Remove.keySet().iterator();
                while(keys.hasNext()){
                    long key = keys.next();
                    if(key < curretTime){
                        QName name = resource2Remove.get(key);
                        //unless resource state has changed,we remove the resource
                        ManagedResource resource = resourceModel.getResource(name);
                        if(resource != null){
                            String state = resource.getState();
                            if(state.equals(SystemState.CrashedState.toString()) 
                                    || state.equals(SystemState.RepairedState.toString())
                                    || state.equals(SystemState.StoppedState.toString())
                                    || state.equals(SystemState.FaultyState.toString())){
                                resourceModel.removeResource(name);
                                context.removeResource(name);  
                                context.getRuleEngine().removeObject(resource);
                                log.info("Meta-Object of the resource "+name + " is removed from Hasthi");
                            }else{
                                log.info("Meta-Object of the resource "+name + " is in " +state+ ",not removed from Hasthi");
                            }
                        }else{
                            log.info("Meta-Object of the resource "+name + " is already removed, probabaly from  rules");
                        }
                        //remove the key
                        keys.remove();
                    }else{
                        //resource2Remove is sorted by the key, so we are done
                        break;
                    }
                }
            }

            /**
             * Following code remove any managers that have not sent heartbeat
             * messages, and reassign the resources assigned to those managers
             */
            Collection<RemoteManagerContext> managers = context.getManagers();
            // Clean Up the managers who failed to send heartbeats
            long managerEpochTime = baseConfigContext.getManagerEpochSeconds() * 5000;
            for (RemoteManagerContext managerContext : managers) {
                if (curretTime - managerContext.getLastUpdated() > managerEpochTime) {
                    // TODO may be we need to cantact the manager and
                    // verify/remove
                    managers2remove.add(managerContext);
                }
            }
            // remove the managers
            if(managers2remove.size() > 0){
                for (RemoteManagerContext manager2Remove : managers2remove) {
                    context.removeManager(manager2Remove.getAddress());
                    dlogger.info("Manager ", manager2Remove.getAddress().getSoapAddress(), " is Down");
                }

                // reassign the resources assigned to removed managers
                for (RemoteManagerContext manager2Remove : managers2remove) {
                    List<QName> resourceNames = manager2Remove.getAssignedResources();
                    for (QName resourceName : resourceNames) {
                        reassignResource(resourceName);
                    }
                }    
            }
       
            // It is possible assigning a resource may fail, we add them to
            // unAssignResources, list
            // and try to reassign those resource in the next epoch
            if(unAssignResources.size() > 0){
                ArrayList<QName> tempUnassignedList = new ArrayList<QName>();
                tempUnassignedList.addAll(unAssignResources);
                unAssignResources.clear();
                for (QName resourceName : unAssignResources) {
                    reassignResource(resourceName);
                }
            }

            /**
             * When the coordinator starts up we give time for other managers
             * and resources to Join before recalculating successors or
             * evaluating rules. Furthermore, if isActiveManagement in the context is set to false, 
             * we do not evaluate the rules
             */
            if (context.isActiveManagement() 
                    //&& (context.getMyManagerContext().isFirstManagerOfRing() || coordinatorAgeSteps > 2)) {
                    && (coordinatorAgeSteps > 3)) {
                List<ManagerAddress> sucessorList = context.createSucessorList();
                //Utils.verifyListIsSorted(sucessorList.toArray(new ManagerAddress[0]));
                /**
                 * Recalculate the sucessors list
                 */
                long point2 = System.nanoTime();
                /**
                 * Run the Global rules
                 */
                // Collection<QName> resources =
                // context.getResourceModel().getResources();
                // long now = System.currentTimeMillis();
                // for(QName name:resources){
                // EditableManagedResource managedResource =
                // context.getResourceModel().getResource(name);
                // managedResource.setLastUpdated(now);
                // //context.getRuleEngine().objectChanged();
                // }
                context.getRuleEngine().fire();

                long point3 = System.nanoTime();

                loc.addAReading(point3 - point1);
                roc.addAReading(point3 - point2);
            }else{
                dlogger.info("Coordiantor Control loop is turned Off");
            }
        } catch (Throwable e) {
            dlogger.errror("Error at Coordinator Deamon " + e.getMessage(), e);
        }
    }

    // calculate successors

    private void reassignResource(QName resourceName) throws HasthiException {
        RemoteManagerContext resourceCaretaker = null;
        try {
            UnicastContianier soapContainer = this.context.getMyManagerContext()
                    .getUnicastContainer();
            String managementEprOfResource = context.getResourceModel().getResource(resourceName)
                    .getManagerAgentURL();

            resourceCaretaker = context.findBestManager4Resource(resourceName);
            ManagerAddress address = resourceCaretaker.getAddress();
            AssignResourcesDocument assignResourceDocument = ManagerClientUtils
                    .createAssignResourceRequest(managementEprOfResource, resourceName);
            MessageContext requestMessage = new SimpleMessageContext(assignResourceDocument,
                    address.getSoapAddress(), ASSIGN_RESOURCE_ACTION);
            AssignResourcesResponseDocument response = (AssignResourcesResponseDocument) soapContainer
                    .sendReceive(requestMessage);
            // TODO do we need to set reassgined resource summery using response
            context.addResource2Manager(resourceName, resourceCaretaker);
        } catch (RemoteProcessorException e) {
            dlogger.errror("Falied to reassign resource", e);
            // TODO there are two cases, we assume resource is down for now
            ManagedResource resource = context.getResourceModel().removeResource(resourceName);
            context.getRuleEngine().removeObject(resource);
        } catch (HasthiException e) {
            dlogger.errror("Manager " + resourceCaretaker != null ? resourceCaretaker.getId() : ""
                    + " is down while trying to assign resource " + resourceName, e);
            unAssignResources.add(resourceName);
            // if
            // (!VerificationTools.veirfyManagerISAvalible(context.getMyManagerContext(),
            // resourceCaretaker.getAddress())) {
            // // TODO for now we will wait for next epoch
            // }
        }

    }

}
